# Copyright (c) 2018 Presto Labs Pte. Ltd.
# Author: jhkim

from typing import List, Union

from coin.base.param_util import to_list
from coin.exchange.base.kr_rest.product import NormProduct
from coin.exchange.base.kr_rest.futures_product import NormFuturesProduct
from coin.exchange.base.feed_arbitration_handler import (HandlerLevelFeedArbHandler,
                                                         MessageLevelFeedArbHandler)
from coin.strategy.mm.feed_builder import FlowBookBuilderBase, _get_feed_group

import coin.exchange.okex.config as okex_config
from coin.exchange.okex_futures.kr_rest.futures_product import OkexFuturesProduct
from coin.exchange.okex_futures.ws.handler import OkexFuturesHandler
from coin.exchange.okex_futures_v3.ws.handler import OkexFuturesHandler as OkexFuturesV3Handler
from coin.exchange.okex_futures_swap.ws.handler import OkexFuturesSwapHandler

from coin.exchange.bitmex.ws.handler import BitmexHandler
from coin.exchange.bitmex.kr_rest.futures_product import BitmexFuturesProduct

import coin.exchange.bitflyer_v1.config as bitflyer_config
import coin.exchange.bitflyer_v1.util as bitflyer_util
from coin.exchange.bitflyer_v1.kr_rest.futures_product import BitflyerFuturesProduct
from coin.exchange.bitflyer_v1.ws.handler import BitflyerHandler

import coin.exchange.huobi.config as huobi_config
from coin.exchange.huobi_futures.kr_rest.futures_product import HuobiFuturesProduct
from coin.exchange.huobi_futures.ws.handler import HuobiFuturesHandler

import coin.exchange.gdax.config as gdax_config
from coin.exchange.gdax.kr_rest.product import GdaxProduct
from coin.exchange.gdax.ws.handler import GdaxHandler

import coin.exchange.binance.config as binance_config
from coin.exchange.binance.kr_rest.product import BinanceProduct
from coin.exchange.binance.ws.snapshot_handler import BinanceSnapshotHandler

import coin.strategy.mm.feed_hash as fhash

from coin.exchange.huobi.kr_rest.product import HuobiProduct
from coin.exchange.huobi.ws.handler import HuobiHandler


class OkexFuturesFeedArbHandler(MessageLevelFeedArbHandler):
  pass


class FeedArbBookBuilder(FlowBookBuilderBase):
  def get_worker_last_feed_ts(self, wid):
    if self._feed_handler is None:
      return None
    return self._feed_handler.get_worker_last_feed_ts(wid)


class OkexFuturesFeedArbFlowBookBuilder(FeedArbBookBuilder):
  ProductType = OkexFuturesProduct
  FeedHandlerType = OkexFuturesFeedArbHandler
  support_book_builder_observer = True

  @staticmethod
  def find_feed_group(product: NormFuturesProduct):
    if isinstance(product, OkexFuturesProduct):
      native_symbol = product.native_subscription_symbol
    else:
      native_symbol = \
          OkexFuturesProduct.FromProduct(product).native_subscription_symbol
    for group, native_symbol_list in okex_config.futures_group_symbol.items():
      if native_symbol in native_symbol_list:
        return group
    raise ValueError(native_symbol)

  @classmethod
  def get_topic_string(cls, product: Union[NormProduct, List[NormProduct], None], worker_id=None):
    groups = _get_feed_group(product, cls.find_feed_group)
    topics = []
    worker_id = to_list(worker_id or ['1', '2'])
    for g in groups:
      for wid in worker_id:
        topics += [
            'okex_ws_futureusd_depth_fullbook_%s%s' % (g, wid),
            'okex_ws_futureusd_depth_diff_%s%s' % (g, wid),
            'okex_ws_futureusd_trade_%s%s' % (g, wid),
            'okex_ws_futureusd_ticker_%s%s' % (g, wid)
        ]
    return topics

  def __init__(self, flow_sub, worker_id=None, feed_checker_config=None, check_limit_price=False):
    super().__init__(feed_checker_config=feed_checker_config)

    for wid in worker_id:
      self._feed_handler.add_handler(wid, OkexFuturesHandler, check_limit_price=check_limit_price)

    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_futureusd_depth_fullbook_*',
                                      'on_first_book',
                                      fhash.get_hash(fhash.okex_feed_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_futureusd_depth_diff_*',
                                      'on_diff',
                                      fhash.get_hash(fhash.okex_diff_feed_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_futureusd_trade_*',
                                      'on_trade',
                                      fhash.get_hash(fhash.data_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_futureusd_ticker_*',
                                      'on_ticker',
                                      fhash.get_hash(fhash.okex_feed_hash, flow_sub))

  def subscribe_book_worker_only(self, *, product, worker, callback):
    if not isinstance(product, OkexFuturesProduct):
      product = OkexFuturesProduct.FromProduct(product)

    native_symbol = product.native_symbol
    self._feed_handler.add_check_product(product)

  def subscribe_book(self, product, callback):
    if not isinstance(product, OkexFuturesProduct):
      product = OkexFuturesProduct.FromProduct(product)

    native_symbol = product.native_symbol
    self._feed_handler.subscribe_book(native_symbol, callback)
    self._feed_handler.add_check_product(product)

  def subscribe_trade(self, product, callback):
    if not isinstance(product, OkexFuturesProduct):
      product = OkexFuturesProduct.FromProduct(product)

    native_symbol = product.native_symbol
    self._feed_handler.subscribe_trade(native_symbol, callback)
    self._feed_handler.add_check_product(product)

  def subscribe_ticker(self, product, callback):
    if not isinstance(product, OkexFuturesProduct):
      product = OkexFuturesProduct.FromProduct(product)

    native_symbol = product.native_symbol
    self._feed_handler.subscribe_event('ticker', native_symbol, callback)


class OkexFuturesV3FeedArbHandler(MessageLevelFeedArbHandler):
  OldFeedTimeOut = 120 * 1e9


class OkexFuturesV3FeedArbFlowBookBuilder(FeedArbBookBuilder):
  ProductType = OkexFuturesProduct
  FeedHandlerType = OkexFuturesV3FeedArbHandler
  support_book_builder_observer = True

  @staticmethod
  def find_feed_group(product: NormFuturesProduct):
    # group_symbol is in v1 (USD quote) and v3 (USDT quote) format.
    if isinstance(product, OkexFuturesProduct):
      native_symbol = product.native_subscription_symbol
      symbol = product.subscription_symbol
    else:
      native_symbol = \
        OkexFuturesProduct.FromProduct(product).native_subscription_symbol
      symbol = OkexFuturesProduct.FromProduct(product).subscription_symbol
    for group, native_symbol_list in okex_config.futures_group_symbol.items():
      if native_symbol in native_symbol_list:
        return group
    for group, symbol_list in okex_config.futures_group_symbol.items():
      if symbol in symbol_list:
        return group
    raise ValueError()

  @classmethod
  def get_topic_string(cls, product: Union[NormProduct, List[NormProduct], None], worker_id=None):
    groups = _get_feed_group(product, cls.find_feed_group)
    topics = []
    worker_id = to_list(worker_id or ['1', '2'])
    for g in groups:
      for wid in worker_id:
        topics += [
            'okex_ws_futures_v3_depth_fullbook_%s%s' % (g, wid),
            'okex_ws_futures_v3_depth_diff_%s%s' % (g, wid),
            'okex_ws_futures_v3_trade_%s%s' % (g, wid),
            'okex_ws_futures_v3_ticker_%s%s' % (g, wid),
            'okex_ws_futures_v3_price_range_%s%s' % (g, wid),
            'okex_ws_futures_v3_mark_price_%s%s' % (g, wid),
            'okex_rest_futures_v3_open_interest_%s%s' % (g, wid),
            # coin2
            'okex_futures_feed_v3_%s%s' % (g, wid),
            'futures_okex_v3_%s%s' % (g, wid)
        ]
    return topics

  def __init__(self, flow_sub, worker_id=None, feed_checker_config=None, check_limit_price=False):
    super().__init__(feed_checker_config=feed_checker_config)

    for wid in worker_id:
      self._feed_handler.add_handler(wid, OkexFuturesV3Handler, check_limit_price=check_limit_price)

    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_futures_v3_depth_fullbook_*',
                                      'on_first_book',
                                      fhash.get_hash(fhash.message_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_futures_v3_depth_diff_*',
                                      'on_diff',
                                      fhash.get_hash(fhash.message_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_futures_v3_trade_*',
                                      'on_trade',
                                      fhash.get_hash(fhash.message_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_futures_v3_ticker_*',
                                      'on_ticker',
                                      fhash.get_hash(fhash.message_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_futures_v3_price_range_*',
                                      'on_price_range',
                                      fhash.get_hash(fhash.message_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_futures_v3_mark_price_*',
                                      'on_mark_price',
                                      fhash.get_hash(fhash.message_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_rest_futures_v3_open_interest_*',
                                      'on_open_interest',
                                      fhash.get_hash(fhash.message_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_futures_feed_v3_*',
                                      'on_coin2',
                                      fhash.get_hash(fhash.message_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'futures_okex_v3_*',
                                      'on_coin2',
                                      fhash.get_hash(fhash.message_hash, flow_sub))

  def subscribe_book_worker_only(self, *, product, worker, callback):
    if not isinstance(product, OkexFuturesProduct):
      product = OkexFuturesProduct.FromProduct(product)

    native_symbol = product.native_symbol_v3
    self._feed_handler.add_check_product(product)

  def subscribe_book(self, product, callback):
    if not isinstance(product, OkexFuturesProduct):
      product = OkexFuturesProduct.FromProduct(product)

    native_symbol = product.native_symbol_v3
    self._feed_handler.subscribe_book(native_symbol, callback)
    self._feed_handler.add_check_product(product)

  def subscribe_trade(self, product, callback):
    if not isinstance(product, OkexFuturesProduct):
      product = OkexFuturesProduct.FromProduct(product)

    native_symbol = product.native_symbol_v3
    self._feed_handler.subscribe_trade(native_symbol, callback)
    self._feed_handler.add_check_product(product)

  def subscribe_ticker(self, product, callback):
    if not isinstance(product, OkexFuturesProduct):
      product = OkexFuturesProduct.FromProduct(product)

    native_symbol = product.native_symbol_v3
    self._feed_handler.subscribe_event('ticker', native_symbol, callback)

  def subscribe_instrument(self, product, callback):
    if not isinstance(product, OkexFuturesProduct):
      product = OkexFuturesProduct.FromProduct(product)

    native_symbol = product.native_symbol_v3
    self._feed_handler.subscribe_event('instrument', native_symbol, callback)


class OkexFuturesSwapFeedArbHandler(MessageLevelFeedArbHandler):
  pass


class OkexFuturesSwapFeedArbFlowBookBuilder(FeedArbBookBuilder):
  ProductType = OkexFuturesProduct
  FeedHandlerType = OkexFuturesSwapFeedArbHandler
  support_book_builder_observer = True

  @staticmethod
  def find_feed_group(product: NormFuturesProduct):
    if isinstance(product, OkexFuturesProduct):
      native_symbol = product.native_symbol_v3
    else:
      native_symbol = \
          OkexFuturesProduct.FromProduct(product).native_symbol_v3
    for group, native_symbol_list in okex_config.swap_group_symbol.items():
      if native_symbol in native_symbol_list:
        return group
    raise ValueError(native_symbol)

  @classmethod
  def get_topic_string(cls, product: Union[NormProduct, List[NormProduct], None], worker_id=None):
    groups = _get_feed_group(product, cls.find_feed_group)
    topics = []
    worker_id = to_list(worker_id or ['1', '2'])
    for g in groups:
      for wid in worker_id:
        topics += [
            'okex_ws_swap_depth_fullbook_%s%s' % (g, wid),
            'okex_ws_swap_depth_diff_%s%s' % (g, wid),
            'okex_ws_swap_trade_%s%s' % (g, wid),
            'okex_ws_swap_ticker_%s%s' % (g, wid),
            'okex_ws_swap_funding_rate_%s%s' % (g, wid),
            'okex_ws_swap_mark_price_%s%s' % (g, wid),
            'okex_rest_swap_open_interest_%s%s' % (g, wid)
        ]
    return topics

  def __init__(self, flow_sub, worker_id=None, feed_checker_config=None):
    super().__init__(feed_checker_config=feed_checker_config)

    for wid in worker_id:
      self._feed_handler.add_handler(wid, OkexFuturesSwapHandler)

    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_swap_depth_fullbook_*',
                                      'on_first_book',
                                      fhash.get_hash(fhash.message_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_swap_depth_diff_*',
                                      'on_diff',
                                      fhash.get_hash(fhash.message_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_swap_trade_*',
                                      'on_trade',
                                      fhash.get_hash(fhash.message_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_swap_ticker_*',
                                      'on_ticker',
                                      fhash.get_hash(fhash.message_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_swap_funding_rate_*',
                                      'on_funding_rate',
                                      fhash.get_hash(fhash.message_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_ws_swap_mark_price_*',
                                      'on_mark_price',
                                      fhash.get_hash(fhash.message_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'okex_rest_swap_open_interest_*',
                                      'on_open_interest',
                                      fhash.get_hash(fhash.message_hash, flow_sub))

  def subscribe_book_worker_only(self, *, product, worker, callback):
    if not isinstance(product, OkexFuturesProduct):
      product = OkexFuturesProduct.FromProduct(product)

    native_symbol = product.native_symbol_v3
    self._feed_handler.add_check_product(product)

  def subscribe_book(self, product, callback):
    if not isinstance(product, OkexFuturesProduct):
      product = OkexFuturesProduct.FromProduct(product)

    native_symbol = product.native_symbol_v3
    self._feed_handler.subscribe_book(native_symbol, callback)
    self._feed_handler.add_check_product(product)

  def subscribe_trade(self, product, callback):
    if not isinstance(product, OkexFuturesProduct):
      product = OkexFuturesProduct.FromProduct(product)

    native_symbol = product.native_symbol_v3
    self._feed_handler.subscribe_trade(native_symbol, callback)
    self._feed_handler.add_check_product(product)

  def subscribe_ticker(self, product, callback):
    if not isinstance(product, OkexFuturesProduct):
      product = OkexFuturesProduct.FromProduct(product)

    native_symbol = product.native_symbol_v3
    self._feed_handler.subscribe_event('ticker', native_symbol, callback)

  def subscribe_instrument(self, product, callback):
    if not isinstance(product, OkexFuturesProduct):
      product = OkexFuturesProduct.FromProduct(product)

    native_symbol = product.native_symbol_v3
    self._feed_handler.subscribe_event('instrument', native_symbol, callback)


class BitmexFeedArbHandler(HandlerLevelFeedArbHandler):
  pass


class BitmexFeedArbFlowBookBuilder(FeedArbBookBuilder):
  ProductType = BitmexFuturesProduct
  FeedHandlerType = BitmexFeedArbHandler

  @staticmethod
  def find_feed_group(product: NormProduct):
    return 'a'

  @classmethod
  def get_topic_string(cls, product: Union[NormProduct, List[NormProduct], None], worker_id=None):
    topics = []
    worker_id = to_list(worker_id or ['1', '2'])
    for wid in worker_id:
      topics += [
          'bitmex_ws_orderBookL2_a%s' % wid,
          'bitmex_ws_trade_a%s' % wid,
          'bitmex_ws_instrument_%s' % wid,
          'bitmex_ws_liquidation_%s' % wid
      ]
    return topics

  def __init__(self, flow_sub, worker_id=None, feed_checker_config=None):
    super().__init__(feed_checker_config=feed_checker_config)

    for wid in worker_id:
      self._feed_handler.add_handler(wid, BitmexHandler)
    self._feed_handler.subscribe_multi_shift(flow_sub,
                                             'bitmex_ws_orderBookL2_*',
                                             'on_ws_orderbook_l2')
    self._feed_handler.subscribe_flow(flow_sub,
                                      'bitmex_ws_trade_*',
                                      'on_ws_trade',
                                      fhash.get_hash(fhash.bitmex_trdmatchid, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'bitmex_ws_instrument_*',
                                      'on_ws_instrument',
                                      fhash.get_hash(fhash.data_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'bitmex_ws_liquidation_*',
                                      'on_ws_liquidation',
                                      fhash.get_hash(fhash.data_hash, flow_sub))

  def subscribe_book(self, product, callback):
    product = BitmexFuturesProduct.FromProduct(product)
    native_symbol = product.native_symbol
    self._feed_handler.subscribe(native_symbol, callback)
    self._feed_handler.add_check_product(product)

  def subscribe_trade(self, product, callback):
    product = BitmexFuturesProduct.FromProduct(product)
    native_symbol = product.native_symbol
    self._feed_handler.subscribe_trade(native_symbol, callback)
    self._feed_handler.add_check_product(product)

  def subscribe_instrument(self, product, callback):
    product = BitmexFuturesProduct.FromProduct(product)
    native_symbol = product.native_symbol
    self._feed_handler.subscribe_instrument(native_symbol, callback)


class BitflyerFeedArbHandler(MessageLevelFeedArbHandler):
  pass


# ----------------------------
# Bitflyer
# ----------------------------
class BitflyerFeedArbFlowBookBuilder(FeedArbBookBuilder):
  ProductType = BitflyerFuturesProduct
  FeedHandlerType = BitflyerFeedArbHandler

  @staticmethod
  def find_feed_group(product: NormFuturesProduct):
    native_symbol = bitflyer_util.get_native_symbol(product)
    for group, native_symbol_list in bitflyer_config.group_symbol.items():
      if native_symbol in native_symbol_list:
        return group
    raise ValueError(native_symbol)

  @classmethod
  def get_topic_string(cls,
                       product: Union[NormFuturesProduct, List[NormFuturesProduct], None],
                       worker_id=None):
    groups = _get_feed_group(product, cls.find_feed_group)
    topics = []
    worker_id = to_list(worker_id or ['1', '2'])
    for wid in worker_id:
      for g in groups:
        topics += [
            'bitflyer_ws_ticker_board_%s%s' % (g, wid),
        ]
    return topics

  def __init__(self, flow_sub, worker_id=None, feed_checker_config=None):
    super().__init__(feed_checker_config=feed_checker_config)

    for wid in worker_id:
      self._feed_handler.add_handler(wid, BitflyerHandler)

    # TODO: fix to use bitmex uid if possible
    self._feed_handler.subscribe_flow(flow_sub,
                                      'bitflyer_ws_ticker_board_*',
                                      'on_message',
                                      fhash.get_hash(fhash.bitflyer_feed_hash, flow_sub))

  def subscribe_book(self, product, callback):
    native_symbol = bitflyer_util.get_native_symbol(product)
    self._feed_handler.subscribe_book(native_symbol, callback)
    self._feed_handler.add_check_product(product)

  def subscribe_trade(self, product, callback):
    native_symbol = bitflyer_util.get_native_symbol(product)
    self._feed_handler.subscribe_trade(native_symbol, callback)
    self._feed_handler.add_check_product(product)


class HuobiFuturesFeedArbHandler(HandlerLevelFeedArbHandler):
  pass


# ----------------------------
# Hbdm
# ----------------------------
class HuobiFuturesFeedArbFlowBookBuilder(FeedArbBookBuilder):
  ProductType = HuobiFuturesProduct
  FeedHandlerType = HuobiFuturesFeedArbHandler

  @staticmethod
  def find_feed_group(product: NormFuturesProduct):
    native_symbol = product.native_subscription_symbol
    for group, native_symbol_list in huobi_config.futures_group_symbol.items():
      if native_symbol in native_symbol_list:
        return group
    raise ValueError(native_symbol)

  @classmethod
  def get_topic_string(cls,
                       product: Union[NormFuturesProduct, List[NormFuturesProduct], None],
                       worker_id=None):
    groups = _get_feed_group(product, cls.find_feed_group)
    topics = []
    worker_id = to_list(worker_id or ['1', '2'])
    for wid in worker_id:
      for g in groups:
        topics += [
            'huobi_ws_futures_depth_%s%s' % (g, wid), 'huobi_ws_futures_trade_%s%s' % (g, wid)
        ]
    return topics

  def __init__(self, flow_sub, worker_id=None, feed_checker_config=None):
    super().__init__(feed_checker_config=feed_checker_config)

    for wid in worker_id:
      self._feed_handler.add_handler(wid, HuobiFuturesHandler)

    self._feed_handler.subscribe_flow(flow_sub,
                                      'huobi_ws_futures_depth_*',
                                      'on_depth',
                                      fhash.get_hash(fhash.huobi_depth_hash, flow_sub))

    self._feed_handler.subscribe_flow(flow_sub,
                                      'huobi_ws_futures_trade_*',
                                      'on_trade',
                                      fhash.get_hash(fhash.huobi_trade_hash, flow_sub))

  def subscribe_book(self, product, callback):
    native_symbol = product.native_symbol
    self._feed_handler.subscribe_book(native_symbol, callback)
    self._feed_handler.add_check_product(product)

  def subscribe_trade(self, product, callback):
    native_symbol = product.native_symbol
    self._feed_handler.subscribe_trade(native_symbol, callback)
    self._feed_handler.add_check_product(product)


class HuobiFeedArbHandler(HandlerLevelFeedArbHandler):
  pass


# ----------------------------
# Huobi Spot
# ----------------------------
class HuobiFeedArbFlowBookBuilder(FeedArbBookBuilder):
  ProductType = HuobiProduct
  FeedHandlerType = HuobiFeedArbHandler

  @staticmethod
  def find_feed_group(product: NormProduct):
    native_symbol = HuobiProduct.FromProduct(product).native_symbol
    for group, native_symbol_list in huobi_config.group_symbol.items():
      if native_symbol in native_symbol_list:
        return group
    raise ValueError(native_symbol)

  @classmethod
  def get_topic_string(cls, product: Union[NormProduct, List[NormProduct], None], worker_id=None):
    groups = _get_feed_group(product, cls.find_feed_group)
    topics = []
    worker_id = to_list(worker_id or ['1', '2'])
    for wid in worker_id:
      for g in groups:
        topics += ['huobi_ws_depth_%s%s' % (g, wid), 'huobi_ws_trade_%s%s' % (g, wid)]
    return topics

  def __init__(self, flow_sub, worker_id=None, feed_checker_config=None):
    super().__init__(feed_checker_config=feed_checker_config)

    for wid in worker_id:
      self._feed_handler.add_handler(wid, HuobiHandler)

    self._feed_handler.subscribe_flow(
        flow_sub,
        'huobi_ws_depth_*',
        'on_ws_book',
        fhash.get_hash(fhash.huobi_spot_depth_hash, flow_sub, robust_json_parser=True))

    self._feed_handler.subscribe_flow(
        flow_sub,
        'huobi_ws_delta_depth_*',
        'on_ws_diff',
        fhash.get_hash(fhash.huobi_spot_delta_depth_hash, flow_sub, robust_json_parser=True))

    self._feed_handler.subscribe_flow(
        flow_sub,
        'huobi_ws_trade_*',
        'on_ws_trade',
        fhash.get_hash(fhash.huobi_spot_trade_hash, flow_sub, robust_json_parser=True))

  def subscribe_book(self, product, callback):
    native_symbol = product.native_symbol
    self._feed_handler.subscribe_book(native_symbol, callback)
    self._feed_handler.add_check_product(product)

  def subscribe_trade(self, product, callback):
    native_symbol = product.native_symbol
    self._feed_handler.subscribe_trade(native_symbol, callback)
    self._feed_handler.add_check_product(product)


class GdaxFeedArbHandler(HandlerLevelFeedArbHandler):
  pass


# ----------------------------
# Gdax
# ----------------------------
class GdaxFeedArbFlowBookBuilder(FeedArbBookBuilder):
  ProductType = GdaxProduct
  FeedHandlerType = GdaxFeedArbHandler

  @staticmethod
  def find_feed_group(product: NormProduct):
    native_symbol = GdaxProduct.FromProduct(product).native_symbol
    for group, native_symbol_list in gdax_config.group_symbol.items():
      if native_symbol in native_symbol_list:
        return group
    raise ValueError(native_symbol)

  @classmethod
  def get_topic_string(cls,
                       product: Union[NormFuturesProduct, List[NormFuturesProduct], None],
                       worker_id=None):
    groups = _get_feed_group(product, cls.find_feed_group)
    topics = []
    worker_id = to_list(worker_id or ['1', '2'])
    for wid in worker_id:
      for g in groups:
        topics += ['gdax_ws_level2_%s%s' % (g, wid), 'gdax_ws_full_%s%s' % (g, wid)]
    return topics

  def __init__(self, flow_sub, worker_id=None, feed_checker_config=None):
    super().__init__(feed_checker_config=feed_checker_config)

    for wid in worker_id:
      self._feed_handler.add_handler(wid, GdaxHandler)

    self._feed_handler.subscribe_flow(flow_sub,
                                      'gdax_ws_level2_*',
                                      'on_book',
                                      fhash.get_hash(fhash.gdax_onbook_hash, flow_sub))
    self._feed_handler.subscribe_flow(flow_sub,
                                      'gdax_ws_full_*',
                                      'on_full',
                                      fhash.get_hash(fhash.gdax_onfull_hash, flow_sub))

  def subscribe_book(self, product, callback):
    native_symbol = product.native_symbol
    self._feed_handler.subscribe_book(native_symbol, callback)
    self._feed_handler.add_check_product(product)

  def subscribe_trade(self, product, callback):
    native_symbol = product.native_symbol
    self._feed_handler.subscribe_trade(native_symbol, callback)
    self._feed_handler.add_check_product(product)


class BinanceSnapshotFeedArbHandler(MessageLevelFeedArbHandler):
  pass


# check with:
# ./pyrunner coin/strategy/mm/tool/feed_archive_checker.py  --trading_date=20190802 --exchange=Binance --market_type=Spot --symbols=BTC-USDT --worker_ids=0 --use_feed_cache=False --feed_machine=feed-01.ap-northeast-1.aws --time_range=0-0.5 &
# ./pyrunner coin/strategy/mm/tool/feed_archive_checker.py  --trading_date=20190802 --exchange=Binance --market_type=Spot --symbols=BTC-USDT --worker_ids=0 --use_feed_cache=False --feed_machine=feed-01.ap-northeast-1.aws --time_range=0-0.5 &
# ./pyrunner coin/strategy/mm/tool/feed_archive_checker.py  --trading_date=20190802 --exchange=Binance --market_type=Spot --symbols=BTC-USDT --worker_ids=2 --use_feed_cache=False --feed_machine=feed-01.ap-northeast-1.aws --time_range=0-0.5 &
# ----------------------------
# Binance
# ----------------------------
class BinanceSnapshotFeedArbFlowBookBuilder(FeedArbBookBuilder):
  ProductType = BinanceProduct
  FeedHandlerType = BinanceSnapshotFeedArbHandler

  @staticmethod
  def find_feed_group(product: NormProduct):
    native_symbol = BinanceProduct.FromProduct(product).native_symbol
    for group, native_symbol_list in binance_config.group_symbol.items():
      if native_symbol in native_symbol_list:
        return group
    raise ValueError(native_symbol)

  @classmethod
  def get_topic_string(cls, product: Union[NormProduct, List[NormProduct], None], worker_id=None):
    groups = _get_feed_group(product, cls.find_feed_group)
    topics = []
    worker_id = to_list(worker_id or ['1', '2'])
    for wid in worker_id:
      for g in groups:
        topics += ['binance_ws_depth20_%s%s' % (g, wid), 'binance_ws_trade_%s%s' % (g, wid)]
    return topics

  def __init__(self, flow_sub, worker_id=None, feed_checker_config=None):
    super().__init__(feed_checker_config=feed_checker_config)

    for wid in worker_id:
      self._feed_handler.add_handler(wid, BinanceSnapshotHandler)

    self._feed_handler.subscribe_flow(flow_sub,
                                      'binance_ws_trade_*',
                                      'on_ws_trade',
                                      fhash.get_hash(fhash.binance_ws_trade_hash, flow_sub))

    self._feed_handler.subscribe_flow(flow_sub,
                                      'binance_ws_depth20_*',
                                      'on_ws_book',
                                      fhash.get_hash(fhash.binance_ws_book20_hash, flow_sub))

  def subscribe_book(self, product, callback):
    native_symbol = product.native_symbol
    self._feed_handler.subscribe_book(native_symbol, callback)
    self._feed_handler.add_check_product(product)

  def subscribe_trade(self, product, callback):
    native_symbol = product.native_symbol
    self._feed_handler.subscribe_trade(native_symbol, callback)
    self._feed_handler.add_check_product(product)
